package cn.doitedu.etl;

import cn.doitedu.utils.TimeTruncUDF2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/17
 * @Desc: 学大数据，上多易教育
 **/
public class Job02_流量概况olap主题聚合宽表 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(4);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射 kafka 的dwd层的topic: dwd-events
        tenv.executeSql("create table dwd_events_kafka(\n" +
                "     release_channel    string\n" +
                "    ,device_type        string\n" +
                "    ,session_id         string\n" +
                "    ,app_version        string\n" +
                "    ,lat                double\n" +
                "    ,lng                double\n" +
                "    ,user_name          string\n" +
                "    ,event_id           string\n" +
                "    ,action_time        bigint\n" +
                "    ,properties         map<string,string>\n" +
                "    ,user_id bigint\n" +
                "    ,member_level_id bigint\n" +
                "    ,nickname string\n" +
                "    ,phone string\n" +
                "    ,status int\n" +
                "    ,create_time timestamp(3)\n" +
                "    ,is_new as  if(date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE,1,0)     \n " +
                "    ,icon string\n" +
                "    ,gender int\n" +
                "    ,birthday date\n" +
                "    ,register_city string\n" +
                "    ,job string\n" +
                "    ,personalized_signature string\n" +
                "    ,source_type int\n" +
                "    ,integration int\n" +
                "    ,growth int\n" +
                "    ,luckey_count int\n" +
                "    ,history_integration int\n" +
                "    ,modify_time timestamp(3)\n" +
                "    ,province   string\n" +
                "    ,city       string\n" +
                "    ,region     string\n" +
                "    ,url as properties['url'] \n" +
                "    ,page_type       string\n" +
                "    ,page_service    string\n" +
                "    ,rt as to_timestamp_ltz(action_time,3) \n" +
                "    ,watermark for rt as rt  \n" +
                "\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-events',\n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "    'properties.group.id' = 'doit44_g1',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'value.format' = 'json',\n" +
                "    'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")                                                    ");


        // 创建doris物理聚合表的连接器映射表
        tenv.executeSql(
                " create table doris_sink(                                                                               "+
                        " dt                   date                                                                              "+
                        " ,m60_time            string                                                                            "+
                        " ,m30_time            string                                                                            "+
                        " ,m10_time            string                                                                            "+
                        " ,m05_time            string                                                                            "+
                        " ,m01_time            string                                                                            "+
                        " ,province            string                                                                            "+
                        " ,city                string                                                                            "+
                        " ,region              string                                                                            "+
                        " ,device_type         string                                                                            "+
                        " ,release_channel     string                                                                            "+
                        " ,app_version         string                                                                            "+
                        " ,url                 string                                                                            "+
                        " ,page_type           string                                                                            "+
                        " ,page_service        string                                                                            "+
                        " ,is_new              int                                                                               "+
                        " ,pv                  bigint                                                                            "+
                        " ,user_id             bigint                                                                            "+
                        " ,session_id          string                                                                            "+
                        "                                                                                                        "+
                        " ) WITH (                                                                                               "+
                        "    'connector' = 'doris',                                                                              "+
                        "    'fenodes' = 'doitedu:8030',                                                                         "+
                        "    'table.identifier' = 'dws.olap_traffic_agg_01',                                                     "+
                        "    'username' = 'root',                                                                                "+
                        "    'password' = 'root',                                                                                "+
                        "    'sink.label-prefix' = 'doris_label-105',                                                            "+
                        "    'sink.properties.columns' = 'dt,m60_time,m30_time,m10_time,m05_time,m01_time,province,city,region,  "+
                        "           device_type,release_channel,app_version,url,page_type,page_service,is_new,                   "+
                        " 		  pv,user_id,session_id,uv_bitmap=to_bitmap(user_id),ses_hll=hll_hash(session_id)'               "+
                        " )                                                                                                      "
        );



        // 按找主题宽表的维度、粒度，聚合 pv
        tenv.createTemporaryFunction("time_trunc", TimeTruncUDF2.class);
        tenv.executeSql(
                        "insert into doris_sink \n" +
                        "with tmp as (\n" +
                        "select  *\n" +
                        "from dwd_events_kafka\n" +
                        "where event_id = 'page_load' )\n" +
                        "\n" +
                        "select\n" +
                        "     to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd'))   as dt \n" +
                        "    ,time_trunc(action_time,60)     as m60_time\n" +
                        "    ,time_trunc(action_time,30)     as m30_time\n" +
                        "    ,time_trunc(action_time,10)     as m10_time\n" +
                        "    ,time_trunc(action_time,5)      as m05_time\n" +
                        "    ,time_trunc(action_time,1)      as m01_time\n" +
                        "    ,province\n" +
                        "    ,city\n" +
                        "    ,region\n" +
                        "    ,device_type\n" +
                        "    ,release_channel\n" +
                        "    ,app_version\n" +
                        "    ,url\n" +
                        "    ,page_type\n" +
                        "    ,page_service\n" +
                        "    ,is_new\n" +
                        "    ,count(1) as pv \n" +
                        "    ,user_id\n" +
                        "    ,session_id\n" +
                        "from table(\n" +
                        "    tumble(table tmp,descriptor(rt),interval '1' minute)\n" +
                        ")\n" +
                        "group by \n" +
                        "    window_start\n" +
                        "    ,window_end\n" +
                        "    ,date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd') \n" +
                        "    ,time_trunc(action_time,60)\n" +
                        "    ,time_trunc(action_time,30)\n" +
                        "    ,time_trunc(action_time,10)\n" +
                        "    ,time_trunc(action_time,5)\n" +
                        "    ,time_trunc(action_time,1)\n" +
                        "    ,province\n" +
                        "    ,city\n" +
                        "    ,region\n" +
                        "    ,device_type\n" +
                        "    ,release_channel\n" +
                        "    ,app_version\n" +
                        "    ,url\n" +
                        "    ,page_type\n" +
                        "    ,page_service\n" +
                        "    ,is_new\n" +
                        "    ,user_id\n" +
                        "    ,session_id").print();
    }
}
